package com.tcoiss.message.config;


import com.tcoiss.message.service.impl.MessageListenerService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * 消息MQ配置
 *
 * @author tcoiss
 */
@Configuration
@Component
public class MQConsumerConfig
{
    public static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerConfig.class);
    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.instanceName}")
    private String instanceName; // 当前消费者实例名称
    @Value("${rocketmq.consumer.consumeThreadMax}")
    private int maxThread;
    @Value("${rocketmq.consumer.consumeThreadMin}")
    private int minThread;
    /*// 订阅指定的 topic
    @Value("#{'${rocketmq.consumer.subscribes}'.split(',')}")
    private List<String> subscribes;*/

    @Value("${rocketmq.consumer.allowRetryTime}")
    private int allowRetryTime; // 允许重试次数
    //@Autowired
    //private IMdmReceiveHandle mdmReceiveHandle; // 处理逻辑类



    @Autowired
    private MessageListenerService messageListenerService;

    @Bean
    @ConditionalOnMissingBean
    public DefaultMQPushConsumer defaultMQPushConsumer() throws RuntimeException {
        // 初始化消费者
        DefaultMQPushConsumer defaultMQConsumer = new DefaultMQPushConsumer(groupName);
        defaultMQConsumer.setNamesrvAddr(namesrvAddr);
        defaultMQConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);// 默认初次初始化从第一条获取消息
        defaultMQConsumer.setConsumeMessageBatchMaxSize(1);// 一次接收一条记录
        defaultMQConsumer.setMessageModel(MessageModel.BROADCASTING);// 使用非集群消费模式
        defaultMQConsumer.setInstanceName(instanceName);
        defaultMQConsumer.setVipChannelEnabled(false);
        if (maxThread != 0 && minThread != 0) {
            defaultMQConsumer.setConsumeThreadMax(maxThread);
            defaultMQConsumer.setConsumeThreadMin(minThread);
        }
        List<String> subscribes = new ArrayList<>();
        try {
            // 设置该消费者订阅的主题和tag，如果是订阅该主题下的所有tag，使用*；
            for (String subscribe : subscribes) {
                LOGGER.info("subscribe Topic,Topic:" + subscribe + ",tags: *");
                defaultMQConsumer.subscribe(subscribe, "*");// 设置订阅条件
            }
            defaultMQConsumer.setPullBatchSize(1);// 设置一次获取消息记录
            defaultMQConsumer.registerMessageListener(messageListenerService);
            // 启动消费
            defaultMQConsumer.start();
            LOGGER.info("consumer is started. groupName:{}, topics:{}, namesrvAddr:{}",groupName,subscribes.toString(),namesrvAddr);

        } catch (Exception e) {
            LOGGER.error("failed to start consumer . groupName:{}, topics:{}, namesrvAddr:{}",groupName,subscribes.toString(),namesrvAddr,e);
            throw new RuntimeException(e);
        }
        return defaultMQConsumer;
    }

}
